package com.wuji1626.spark.core.wc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}


object Sample02_WordCount {
  def main(args: Array[String]): Unit = {

    // Application Spark Framework
    // 建立与 Spark 框架的链接
    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(sparkConf)
    // 执行业务操作
    // 1.读取文件，读取一行行的数据
    // hello world
    val lines: RDD[String] = sc.textFile(path = "datas")

    // 2.将一行数据进行拆分，形成一个个的单词（分词），打平
    // "hello world" => hello, world, hello, world
    val words: RDD[String] = lines.flatMap(_.split(" "))

    // 3.将数据根据单词直接变成元组，（hello，1）（world，1）
    //
    val word2One = words.map(
      word => (word, 1)
    )

    // 4.对分组后的数据进行转换
    // (hello, 3),(world,2)
    val wordGroup: RDD[(String, Iterable[(String, Int)])] = word2One.groupBy(
      t => t._1
    )
    var wordCount = wordGroup.map{
      case (word, list)=> {
        list.reduce(
          (t1, t2) => {
            (t1._1, t1._2 + t2._2)
          }
        )
      }
    }

    // 5.将转换结果进行打印
    val array: Array[(String, Int)] = wordCount.collect()
    array.foreach(println)

    // 关闭 Spark
    sc.stop()

  }
}

